package cn.jly.bigdata.spark.sql

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

/**
 * @author lanyangji
 * @date 2019/11/30 15:56
 */
object SparkSQL02_Transform {

  def main(args: Array[String]): Unit = {

    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkSQL02_Transform")

    val spark: SparkSession = SparkSession.builder().config(sparkConf).getOrCreate()

    // 需要引入隐式转换，这里的spark指的是SparkSession对象的名字
    import spark.implicits._
    // --- 以上这三步，尽量以后都要加上-----

    // 创建RDD
    val dataRdd: RDD[(Int, String, Int)] = spark.sparkContext.makeRDD(List((1, "zhangsan", 22), (2, "lisi", 33), (3, "wangwu", 44)))

    // --------------------------------------------
    // 转换为DF
    //    val df: DataFrame = dataRdd.toDF()
    // 要不传参数，传参数的话，数量要和数据一致
    val df: DataFrame = dataRdd.toDF("id", "name", "age")
    df.show()

    // --------------------------------------------
    // 转换为DS
    // 增加样例类 User
    val ds: Dataset[Employee] = df.as[Employee]
    ds.show()

    // --------------------------------------------
    // 转换为DF
    val df1: DataFrame = ds.toDF()

    // --------------------------------------------
    // 转换为RDD
    val rdd1: RDD[Row] = df1.rdd

    rdd1.foreach(row => {
      // 获取数据时，可以通过索引来访问数据
      println(row.getInt(0))
      println(row.getString(1))
      println(row.getInt(2))
    })

    // -------------rdd -> ds -  rdd + 结构 + 类 = rdd + 样例类 = DS, rdd + 结构 = DF
    val userRdd: RDD[User] = dataRdd.map {
      case (id, name, age) => User(id, name, age)
    }
    val userDS: Dataset[User] = userRdd.toDS()
    userDS.show()

    val rdd2: RDD[User] = userDS.rdd
    rdd2.foreach(println)

    // 释放资源
    spark.close()
  }
}

case class User(id: Int, name: String, age: Int)